package org.tigger

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：触发器
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/24 19:47
 * modified By ：
 * version:    : 1.0
 */
object TiggerExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 处理数据
    source
      .assignAscendingTimestamps(_.ts)
      .keyBy(_.id)
      .timeWindow(Time.seconds(5))
      .trigger(new MyTrigger)
      .process(new MyProcess)
      .print()
  }

  class MyTrigger extends Trigger[SensorReading, TimeWindow] {

    // 每个元素执行一次
    override def onElement(element: SensorReading, timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = {
      // 第一次触发，进来之后就创建一触发器
      val isFrist = ctx.getPartitionedState(
        new ValueStateDescriptor[Boolean]("trigger", Types.of[Boolean])
      )

      // 创建两个触发器，一个是窗口关闭时间，一个是第一个整秒定时器
      if (!isFrist.value()) {
        // 设置窗口关闭时间定时器
        ctx.registerEventTimeTimer(window.getEnd)

        // 设置第一个整秒定时器
        val nextTime = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
        ctx.registerEventTimeTimer(nextTime)

        // 将 isFrist 设置为 true
        isFrist.update(true)
      }
      TriggerResult.CONTINUE
    }

    // 被系统触发时间定时器触发，什么都不做
    override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = TriggerResult.CONTINUE

    // 定时器被触发
    override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext) = {
      println("******************" + time)
      if (time == window.getEnd) { // 窗口关闭时间
        TriggerResult.FIRE_AND_PURGE
      } else { // 整秒定时器
        val nextTime = ctx.getCurrentWatermark + (1000 - (ctx.getCurrentWatermark % 1000))
        if (nextTime < window.getEnd) {
          // 注册下一个整秒定时器
          ctx.registerEventTimeTimer(nextTime)
        }
        TriggerResult.FIRE
      }
    }

    // 清除工作
    override def clear(window: TimeWindow, ctx: Trigger.TriggerContext) = {
      val isFrist = ctx.getPartitionedState(
        new ValueStateDescriptor[Boolean]("trigger", Types.of[Boolean])
      )
      isFrist.clear()
    }
  }

  // IN, OUT, KEY, W <: Window>
  class MyProcess extends ProcessWindowFunction[SensorReading, String, String, TimeWindow] {
    override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[String]) = {
      out.collect("触发器触发，当前窗口元素数量是： " + elements.size.toString + " 传感器id是： " + key)
    }
  }

}
